package com.hdaccp.log

import org.apache.spark.sql.{SaveMode, SparkSession}

object Demo1 {
  def main(args: Array[String]): Unit = {
    //得到spark对象
    val spark = SparkSession.builder().appName("SparkStatCleanJob")
      .config("spark.sql.parquet.compression.codec","gzip")
      .master("local[2]").getOrCreate()

    val accessRDD = spark.sparkContext.textFile("F:/accp教学/sparkresources/log4")
    //accessRDD.take(10).foreach(println)
    //RDD ==> DF
   val accessDF = spark.createDataFrame(accessRDD.map(x => AccessConvertUtil.parseLog(x)),
      AccessConvertUtil.struct)

    //    accessDF.printSchema()
    //    accessDF.show(false)

    accessDF.coalesce(1).write.format("parquet").mode(SaveMode.Overwrite)
      .partitionBy("day").save("F:/accp教学/sparkresources/clean5")

    spark.stop
  }
}
